package org.apache.rocketmq.client.codecoord.producer;

import org.apache.rocketmq.client.codecoord.RocketMqConstant;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

/**
 * 单向消息发送
 *
 * @author tianxincoord@163.com
 * @since 2021/10/25
 */
public class BasicOneWayMessageSend {

    public static void main(String[] args) throws Exception {
        ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(10, new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "生产者");
            }
        });
        DefaultMQProducer producer = new DefaultMQProducer(RocketMqConstant.GROUP);
        producer.setNamesrvAddr(RocketMqConstant.NAME_SRV_ADDRESS);
        producer.start();
        producer.setRetryTimesWhenSendFailed(0);

        BasicOneWayMessageSend messageSend = new BasicOneWayMessageSend();
        executor.scheduleWithFixedDelay(() -> {
            try {
                messageSend.oneWayProducer(producer);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, 0, 1, TimeUnit.SECONDS);
    }

    public void oneWayProducer(DefaultMQProducer producer) throws Exception {
        long timeMillis = System.currentTimeMillis();
        String message = "单向消息 " + timeMillis;
        Message msg = new Message(RocketMqConstant.TOPIC, RocketMqConstant.TAG, message.getBytes(RemotingHelper.DEFAULT_CHARSET));
        msg.setKeys(String.valueOf(timeMillis));
        producer.sendOneway(msg);
        System.out.printf("单向消息【%s】已发送%n", message);
    }
}
